<?php
/**
 * 从 Redis Stream 队列消费任务
 * 类似 consumer.php 的简单风格
 */

// 连接 Redis
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);
$redis->select(1);

// 配置
$taskStream = 'task:stream:queue';
$callbackStream = 'task:callback:queue';
$groupName = 'task_processor_group';
$consumerName = 'worker_' . uniqid();

echo "=== 从 Stream 队列消费任务 ===\n";
echo "消费者组: {$groupName}\n";
echo "消费者名称: {$consumerName}\n";
echo "任务队列: {$taskStream}\n";
echo "回调队列: {$callbackStream}\n";
echo "启动时间: " . date('Y-m-d H:i:s') . "\n\n";

// 确保消费者组存在
try {
    $redis->xGroup('CREATE', $taskStream, $groupName, '0', true);
    echo "✓ 消费者组创建成功\n\n";
} catch (RedisException $e) {
    if (strpos($e->getMessage(), 'BUSYGROUP') !== false) {
        echo "✓ 消费者组已存在\n\n";
    } else {
        echo "✗ 创建消费者组失败: " . $e->getMessage() . "\n";
        exit(1);
    }
}

$cycle = 0;
$totalProcessed = 0;

while (true) {
    $cycle++;
    
    try {
        echo "[" . date('Y-m-d H:i:s') . "] 循环 {$cycle} - 开始处理...\n";
        
        // 读取任务
        $tasks = $redis->xReadGroup($groupName, $consumerName, [$taskStream => '>'], 1, 1000);
        
        if (!empty($tasks) && !empty($tasks[$taskStream])) {
            foreach ($tasks[$taskStream] as $messageId => $fields) {
                $taskId = $fields['task_id'] ?? '';
                $taskType = $fields['task_type'] ?? '';
                $taskData = json_decode($fields['task_data'] ?? '{}', true);
                
                echo "处理任务: {$taskId} (类型: {$taskType})\n";
                
                try {
                    // 处理任务
                    $result = processTask($taskType, $taskData, $taskId);
                    
                    // 发送回调
                    $callbackData = [
                        'task_id' => $taskId,
                        'result' => json_encode($result, JSON_UNESCAPED_UNICODE),
                        'status' => 'completed',
                        'processed_at' => time()
                    ];
                    
                    $redis->xAdd($callbackStream, '*', $callbackData);
                    echo "✓ 任务处理完成，回调已发送\n";
                    
                    // 确认任务已处理
                    $redis->xAck($taskStream, $groupName, [$messageId]);
                    echo "✓ 任务已确认\n";

                    
                    // 删除已处理的任务
                    $deletedCount = $redis->xDel($taskStream, [$messageId]);
                    echo "已删除任务: $messageId (删除数量: $deletedCount)\n";
                    
                    $totalProcessed++;
                    
                } catch (Exception $e) {
                    echo "✗ 任务处理失败: " . $e->getMessage() . "\n";
                    
                    // 发送失败回调
                    $errorResult = [
                        'error' => $e->getMessage(),
                        'error_code' => $e->getCode()
                    ];
                    
                    $callbackData = [
                        'task_id' => $taskId,
                        'result' => json_encode($errorResult, JSON_UNESCAPED_UNICODE),
                        'status' => 'failed',
                        'processed_at' => time()
                    ];
                    
                    $redis->xAdd($callbackStream, '*', $callbackData);
                    $redis->xAck($taskStream, $groupName, [$messageId]);
                    
                    echo "✓ 失败回调已发送\n";
                }
                
                echo "\n";
            }
        } else {
            echo "暂无任务需要处理\n";
        }
        
        // 显示统计信息
        if ($cycle % 10 === 0) {
            echo "\n=== 统计信息 (循环 {$cycle}) ===\n";
            echo "总处理任务数: {$totalProcessed}\n";
            
            // 获取队列统计
            $streamInfo = $redis->xInfo('STREAM', $taskStream);
            $callbackInfo = $redis->xInfo('STREAM', $callbackStream);
            echo "任务队列长度: {$streamInfo['length']}\n";
            echo "回调队列长度: {$callbackInfo['length']}\n";
            echo "========================\n\n";
        }
        
        // 休眠
        usleep(500000); // 0.5秒
        
    } catch (Exception $e) {
        echo "错误: " . $e->getMessage() . "\n";
        usleep(1000000); // 1秒
    }
}

/**
 * 处理任务的具体逻辑
 */
function processTask($taskType, $taskData, $taskId) {
    echo "执行任务处理: {$taskType}\n";
    
    switch ($taskType) {
        case 'data_process':
            return processDataProcess($taskData);
            
        case 'file_process':
            return processFileProcess($taskData);
            
        case 'email_send':
            return processEmailSend($taskData);
            
        default:
            throw new Exception("未知的任务类型: {$taskType}");
    }
}

function processDataProcess($data) {
    $input = $data['input'] ?? [];
    $processType = $data['process_type'] ?? 'default';
    
    $result = [];
    switch ($processType) {
        case 'sum':
            $result = array_sum($input);
            break;
        case 'count':
            $result = count($input);
            break;
        case 'filter':
            $result = array_filter($input, function ($item) {
                return !empty($item);
            });
            break;
        case 'sort':
            sort($input);
            $result = $input;
            break;
        case 'transform':
            $result = array_map(function ($item) {
                return strtoupper((string)$item);
            }, $input);
            break;
        default:
            $result = $input;
    }
    
    return [
        'processed_count' => is_array($result) ? count($result) : 1,
        'result' => $result,
        'process_type' => $processType,
        'processed_at' => time()
    ];
}

function processFileProcess($data) {
    $filePath = $data['file_path'] ?? '';
    $operation = $data['operation'] ?? 'read';
    
    if (empty($filePath)) {
        throw new Exception("文件路径不能为空");
    }
    
    switch ($operation) {
        case 'write':
            $content = $data['content'] ?? '';
            file_put_contents($filePath, $content);
            return [
                'file_path' => $filePath,
                'written_bytes' => strlen($content),
                'operation' => 'write',
                'processed_at' => time()
            ];
            
        default:
            throw new Exception("不支持的文件操作: {$operation}");
    }
}

function processEmailSend($data) {
    $to = $data['to'] ?? '';
    $subject = $data['subject'] ?? '';
    $content = $data['content'] ?? '';
    
    if (empty($to) || empty($subject) || empty($content)) {
        throw new Exception("邮件参数不完整");
    }
    
    // 模拟邮件发送
    return [
        'to' => $to,
        'subject' => $subject,
        'content_length' => strlen($content),
        'sent_at' => time(),
        'status' => 'sent'
    ];
}